package com.atguigu.realtime.ads

import java.time.LocalDateTime

import com.atguigu.realtime.BaseAppV4
import com.atguigu.realtime.bean.OrderWide
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.OffsetRange
import org.json4s.jackson.JsonMethods
import scalikejdbc.config.DBs
import scalikejdbc.{DB, SQL}

import scala.collection.mutable.ListBuffer

/**
 * Author atguigu
 * Date 2020/11/24 9:33
 */
object AdsOrderWideApp extends BaseAppV4 {
    override val master: String = "local[2]"
    override val appName: String = "AdsOrderWideApp"
    override val groupId: String = "AdsOrderWideApp"
    override val topic: String = "dws_order_wide"
    override val bachTime: Int = 5
    
    override def run(ssc: StreamingContext,
                     sourceStream: DStream[String],
                     offsetRanges: ListBuffer[OffsetRange]): Unit = {
        DBs.setup()
        sourceStream
            .map(str => {
                implicit val f = org.json4s.DefaultFormats
                val orderWide = JsonMethods.parse(str).extract[OrderWide]
                ((orderWide.tm_id -> orderWide.tm_name), orderWide.final_detail_amount)
            })
            .reduceByKey(_ + _)
            .foreachRDD(rdd => {
                //......
                val now: LocalDateTime = LocalDateTime.now()
                val dateTime = s"${now.toLocalDate} ${now.toLocalTime.toString.substring(0, 8)}"
                val tmAndAmount = rdd
                    .collect()
                    .map {
                        case ((tm_id, tm_name), amount) =>
                            Seq(dateTime, tm_id, tm_name, amount)
                    }
                
                val adsOffset = offsetRanges.map(offsetRange => {
                    Seq(groupId, topic, offsetRange.partition, offsetRange.untilOffset)
                })
                
                DB.localTx(implicit session => {
                    // 这里的代码都会在一个事务中执行
                    // 写数据到mysql
                    val dataSql =
                    """
                      |insert into tm_amount values(?, ?, ?, ?)
                      |""".stripMargin
                    SQL(dataSql).batch(tmAndAmount: _*).apply()
                    //throw new UnsupportedOperationException
                    // 写offset到Mysql
                    val offsetSql =
                    """
                      |replace into ads_offset values(?, ?, ?, ?)
                      |""".stripMargin
                    SQL(offsetSql).batch(adsOffset: _*).apply()
                    println("xxxxxx")
                })
                
            })
    }
}

/*
offset保存到Mysql中, 表的设计
use gmall_result;
CREATE TABLE `ads_offset` (
  `group_id` varchar(200) NOT NULL,
  `topic` varchar(200) NOT NULL,
  `partition_id` int(11) NOT NULL,
  `partition_offset` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`group_id`,`topic`,`partition_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8



 */
